package com.hzh.flink.TF

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo3FlatMapFunction {
  def main(args: Array[String]): Unit = {
    /**
     * 1、创建环境
     */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")

    //javaAPI
    val DS: DataStream[String] = studentDS.flatMap(new FlatMapFunction[String, String] {
      override def flatMap(value: String, out: Collector[String]): Unit = {
        val splits: Array[String] = value.split(",")
        for (words <- splits) {
          //将数据发送到下游
          out.collect(words)
        }
      }
    })

    DS.print()

    env.execute()

  }
}
